package com.lg.monitor

import java.sql.{Connection, DriverManager}

import com.lg.bean.BusInfo
import org.apache.spark.sql.ForeachWriter
object MysqlWritter{
  def getConnection: Connection ={
    Class.forName("com.mysql.jdbc.Driver")
    //获取mysql连接
    DriverManager.getConnection("jdbc:mysql://linux122:3306/test", "root", "12345678")
  }
}
class MysqlWritter extends ForeachWriter[BusInfo]{
  var conn:Connection = _
  override def open(partitionId: Long, epochId: Long): Boolean = {
    conn = MysqlWritter.getConnection
    ture
  }

  override def process(value: BusInfo): Unit = {
    val lglat: Array[String] = value.lglat.split("_")
    var sql = "insert into lat_lgt_info(lgt,lat) values("+lglat(0)+","+lglat(1)+")"
    conn.prepareStatement(sql).executeUpdate()
  }

  override def close(errorOrNull: Throwable): Unit = {
    conn.close()
  }
}
